package main

import (
	"fmt"
	"log"
	"os"
	"os/signal"
	"sync"
	"time"

	"github.com/IBM/sarama"
)

func main() {
	// SendAsyncMessage()
	// GoroutineProcessSend()
	// ConsumeMessageWithZeroOffset()
	ConsumeMessage()
}

func ConsumeMessage() {
	consumer, err := sarama.NewConsumer([]string{"47.98.239.196:9096", "47.98.239.197:9092", "47.98.239.198:9094"}, nil)
	if err != nil {
		panic(err)
	}
	defer func() {
		if err := consumer.Close(); err != nil {
			log.Fatalln(err)
		}
	}()

	topic := "async-topic"
	partitions, err := consumer.Partitions(topic)
	if err != nil {
		log.Fatalln(err)
	}
	consumed := 0
	for _, partition := range partitions {
		partitionConsumer, err := consumer.ConsumePartition(topic, partition, sarama.OffsetOldest)
		if err != nil {
			log.Fatalln(err)
		}
		go func(pc sarama.PartitionConsumer) {
			defer func() {
				if err := pc.Close(); err != nil {
					log.Fatalln(err)
				}
			}()
			for msg := range pc.Messages() {
				consumed++
				fmt.Printf("partition: %d,  offset %d, consumed message: %s\n", partition, msg.Offset, string(msg.Value))
				time.Sleep(500 * time.Millisecond)
			}
		}(partitionConsumer)
	}
	signals := make(chan os.Signal, 1)
	signal.Notify(signals, os.Interrupt, os.Kill)
	select {
	case <-signals:
	}
	log.Printf("Consumed: %d\n", consumed)

}

func ConsumeMessageWithZeroOffset() {
	consumer, err := sarama.NewConsumer([]string{"47.98.239.196:9096", "47.98.239.197:9092", "47.98.239.198:9094"}, nil)
	if err != nil {
		panic(err)
	}
	defer func() {
		if err := consumer.Close(); err != nil {
			log.Fatalln(err)
		}
	}()

	partitionConsumer, err := consumer.ConsumePartition("async-topic", 0, sarama.OffsetOldest)
	if err != nil {
		log.Fatalln(err)
	}

	defer func() {
		if err := partitionConsumer.Close(); err != nil {
			log.Fatalln(err)
		}
	}()

	signals := make(chan os.Signal, 1)
	signal.Notify(signals, os.Interrupt, os.Kill)

	consumed := 0
ConsumerLoop:
	for {
		select {
		case msg := <-partitionConsumer.Messages():
			consumed++
			fmt.Printf("offset %d, consumed message: %s\n", msg.Offset, string(msg.Value))
		case <-signals:
			break ConsumerLoop
		}
		time.Sleep(500 * time.Millisecond)
	}
	log.Printf("Consumed: %d\n", consumed)
}

func GoroutineProcessSend() {
	config := sarama.NewConfig()
	config.Producer.Return.Errors = true
	config.Producer.Return.Successes = true
	config.Producer.Partitioner = sarama.NewRandomPartitioner
	producer, err := sarama.NewAsyncProducer([]string{"47.98.239.196:9096", "47.98.239.197:9092", "47.98.239.198:9094"}, config)
	if err != nil {
		panic(err)
	}

	var wg sync.WaitGroup
	var enqueued, successes, producerErrors int

	wg.Add(1)
	go func() {
		defer wg.Done()
		for suc := range producer.Successes() {
			successes++
			fmt.Println("offset: ", suc.Offset, "partition: ", suc.Partition)
		}
	}()

	wg.Add(1)
	go func() {
		defer wg.Done()
		for err := range producer.Errors() {
			fmt.Println("error:", err)
			producerErrors++
		}
	}()

	signals := make(chan os.Signal, 1)
	signal.Notify(signals, os.Interrupt, os.Kill)
ProducerLoop:
	for {
		select {
		case producer.Input() <- &sarama.ProducerMessage{
			Topic: "my-topic",
			Value: sarama.StringEncoder("hello kafka"),
		}:
			enqueued++
		case <-signals:
			producer.AsyncClose()
			break ProducerLoop
		}
		time.Sleep(500 * time.Millisecond)
	}
}

func SendAsyncMessage() {
	conf := sarama.NewConfig()
	conf.Producer.Return.Successes = true
	producer, err := sarama.NewAsyncProducer([]string{"47.98.239.196:9096", "47.98.239.197:9092", "47.98.239.198:9094"}, conf)
	if err != nil {
		panic(err)
	}
	defer func() {
		if err := producer.Close(); err != nil {
			panic(err)
		}
	}()

	signals := make(chan os.Signal, 1)
	signal.Notify(signals, os.Interrupt, os.Kill)

	var enqueued, successes, producerErrors int
ProducerLoop:
	for {
		select {
		case producer.Input() <- &sarama.ProducerMessage{
			Topic: "async-topic",
			Value: sarama.StringEncoder("hello kafka"),
		}:
			enqueued++
		case err := <-producer.Errors():
			log.Println("Failed to send message", err)
			producerErrors++
		case suc := <-producer.Successes():
			successes++
			fmt.Println("offset: ", suc.Offset, "partition: ", suc.Partition)
		case <-signals:
			break ProducerLoop
		}
		time.Sleep(500 * time.Millisecond)
	}
	log.Printf("Enqueued: %d; errors: %d\n; successes: %d\n", enqueued, producerErrors, successes)
}

func SendSyncMessage() {
	producer, err := sarama.NewSyncProducer([]string{"47.98.239.196:9096", "47.98.239.197:9092", "47.98.239.198:9094"}, nil)
	if err != nil {
		panic(err)
	}
	defer producer.Close()

	msg := &sarama.ProducerMessage{
		Topic: "sync-topic",
		Value: sarama.StringEncoder("hello kafka"),
	}

	partition, offset, err := producer.SendMessage(msg)
	if err != nil {
		log.Printf("failed to send message: %s", err)
		return
	}
	println("Message sent to partition", partition, "at offset", offset)
}
